// optimistic_test.cpp : used to test rollback and GVT

//Two processes: the first one (A) execute one msg and send another msg;
//the second process (B) execute one msg and then receive A's msg with time stamp
//earlier than its current timestamp.

#include "stdafx.h"
#include "mpi.h"
#include <iostream>
#include <fstream>
#include <string>
#include <queue>
#include "Event.h"
#include "global.h"
//#include "EventQueue.h"

using namespace std;

int _tmain(int argc, char** argv)
{
	int rank, size;
	MPI_Init(&argc, &argv);
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &size);
	//cout<<"size"<<size<<"rank"<<rank;

	//creat structure for msg passing, {EventId, time, speed, station, duration, position}
	MPI_Datatype mpi_InitType;
	int array_of_blocklengths[6] = {1,1,1,1,1,1};
	MPI_Aint array_of_displacements[6];
	MPI_Aint addr1, addr2, addr3, addr4, addr5, addr6;
	Init_paras ips;
	MPI_Get_address(&ips.id, &addr1);
	array_of_displacements[0] = 0;
	MPI_Get_address(&ips.time, &addr2);
	array_of_displacements[1] = addr2 - addr1;
	MPI_Get_address(&ips.speed, &addr3);
	array_of_displacements[2] = addr3 - addr1;
	MPI_Get_address(&ips.station, &addr4);
	array_of_displacements[3] = addr4- addr1;
	MPI_Get_address(&ips.duration, &addr5);
	array_of_displacements[4] = addr5 - addr1;
	MPI_Get_address(&ips.position, &addr6);
	array_of_displacements[5] = addr6 - addr1;
	MPI_Datatype array_of_types[6] = {MPI_INT, MPI_DOUBLE, MPI_DOUBLE, MPI_INT, MPI_DOUBLE, MPI_DOUBLE};
	MPI_Type_create_struct(6,array_of_blocklengths,array_of_displacements,array_of_types,&mpi_InitType);
	MPI_Type_commit(&mpi_InitType);
	
	if(rank == 0)//master process
	{
		//deterministic
		int sample = 20;
		
		for(int i=0; i<6; i++)
			cout<<array_of_displacements[i]<<endl;

		//pending event list
		ifstream myfile("C:\\Users\\Amelie-zhou\\Documents\\Visual Studio 2010\\Projects\\optimistic_test\\Debug\\PCS_TEST_DETERMINSTIC.txt");
		if(myfile.is_open())
		{
			cout<<"opend!"<<endl;
			string line;
			getline(myfile,line);
			for(int i=0; i<sample; i++)
			{
				double t,v,d,p;
				int s;
				getline(myfile,line);
				char pattern = ',';
				int pos1 = line.find(pattern, 0);
				int pos2 = line.find(pattern, pos1 + 1);
				int pos3 = line.find(pattern, pos2 + 1);
				int pos4 = line.find(pattern, pos3 + 1);
				string s1 = line.substr(pos1 + 1,pos2 - pos1 - 1);
				t = stod(s1);
				string s2 = line.substr(pos2 + 1, pos3 - pos2 - 1);
				s = stoi(s2) - 1;//start from 0
				string s3 = line.substr(pos3 + 1, pos4 - pos3 - 1);
				d = stod(s3);
				string s4 = line.substr(pos4 + 1, line.length() - pos4 - 1);
				v = stod(s4);
				p = s * 2 + 1;//initialized at the centre of base stations
				Init_paras ps;
				ps.id = Initiation; ps.duration =d; ps.position = p; ps.speed = v; ps.station = s; ps.time = t;
				MPI_Send(&ps,1,mpi_InitType,s+1,0,MPI_COMM_WORLD);			
			}
			Init_paras ps_end;
			ps_end.id = Termination; ps_end.duration =0; ps_end.position = 0; ps_end.speed = 0; ps_end.station = 100; ps_end.time = 0;
			//MPI_Bcast(&ps_end,1,mpi_InitType,0,MPI_COMM_WORLD);
			for(int i=rank+1;i<size;i++)
				MPI_Send(&ps_end,1,mpi_InitType,i,0,MPI_COMM_WORLD);

			//cout<<"end of send"<<endl;
		}
	}

	if(rank != 0)
	{
		vector<Init_paras> init_events;
		while(1)
		{
			//int count;
			Init_paras ps;
			MPI_Status status;
			MPI_Recv(&ps,1,mpi_InitType,0,0,MPI_COMM_WORLD,&status);
			//MPI_Get_count(&status, mpi_InitType, &count);
			//cout << "Rec: " << count << endl;
			//if(count == 1){
			//cout <<rank<<": "<< ps.station << endl; //rank = station+1
			if(ps.station == 100) 
			{
				//cout<<rank<<"breaked!"<<endl;
				break; //received all the initial events
			}
			init_events.push_back(ps);
			//}
		}

		//all initial events pushed in queue
		while(!init_events.empty())
		{
			Init_paras init = init_events.back();
			Event* init_event = new Event(init.time,init.speed,init.station,init.duration,init.position);
			localQue[rank].push(init_event);
			//std::cout<<"this is from process: "<<rank<<endl;
			init_events.pop_back();
		}

		Init_paras in_buffer[100]; //buffer to save input msg
		Init_paras out_buffer[100]; //buffer to save output msg
		queue<state> state_que;
		double curr_time = 0;
		while(!localQue[rank].empty())
		{
			//store state before event handling, timestamp, dropCalls, blockedCalls, totalCalls, Cell
			state curr_state;
			curr_state.time = curr_time; curr_state.blocked = BlockedCalls[rank-1]; 
			curr_state.dropped = droppedCalls[rank-1]; curr_state.total = TotalCalls[rank-1];

			//handle events
			Event* curr_event = localQue[rank].top();			
			if(curr_event->time > curr_time)
			{
				//roll back
				cout<<"pls roll back, process"<<rank<<", curr time is: "<<curr_time<<", next time is: "<<curr_event->time<<endl;
				//send anti-message
				//copy back state
			}

			if(curr_event->id == Termination)
			{
				Cell[curr_event->station] += 1;
			}
			else if(curr_event->id == Initiation)
			{
				if(Cell[curr_event->station] > 0)
				{
					Cell[curr_event->station] -= 1;
					//position = station * 2;
					double endPosition = curr_event->position + curr_event->speed * curr_event->duration / 3600;//km per h
					int endStation = endPosition / 2;
					double travelledDuration = ((curr_event->station + 1)*2 - curr_event->position) / curr_event->speed * 3600;//sec
					if ( curr_event->station == endStation)
					{
						Init_paras ter;
						ter.id = Termination; ter.time = curr_event->time + curr_event->duration; ter.station = endStation;
						ter.duration = ter.position = ter.speed = 0;
						MPI_Send(&ter,1,mpi_InitType,endStation+1,0,MPI_COMM_WORLD);
					}
					else if (curr_event->station+1 > 19)
					{
						Event* ter_event = new Event(curr_event->time + travelledDuration, curr_event->station );
						localQue[curr_event->station+1].push(ter_event); //only for itself, so no need to send/recv
					}
					else
					{
						Init_paras ho;
						ho.id = HandOver; ho.time = curr_event->time + travelledDuration; ho.station = curr_event->station + 1;
						ho.duration = curr_event->duration - travelledDuration; ho.position = 0; ho.speed = curr_event->speed;
						MPI_Send(&ho,1,mpi_InitType,curr_event->station+2,0,MPI_COMM_WORLD);
					}
				}
				else
					BlockedCalls[rank-1] += 1;
				TotalCalls[rank-1] += 1;
			}
			else if(curr_event->id == HandOver)
			{
				Cell[curr_event->station - 1] +=1;
	
				if(Cell[curr_event->station] > 0)
				{
					Cell[curr_event->station] -= 1;
					double position = curr_event->station * 2;
					double endPosition = position + curr_event->speed * curr_event->duration / 3600;
					int endStation = endPosition / 2;
					double travelledDuration = ((curr_event->station + 1)*2 - position) / curr_event->speed * 3600;
					if ( curr_event->station == endStation)
					{
						Init_paras ter;
						ter.id = Termination; ter.time = curr_event->time + curr_event->duration; ter.station = endStation;
						ter.duration = ter.position = ter.speed = 0;
						MPI_Send(&ter,1,mpi_InitType,endStation+1,0,MPI_COMM_WORLD);
					}			
					else if (curr_event->station +1> 19)
					{
						Event* ter_event = new Event(curr_event->time + travelledDuration, curr_event->station );
						localQue[curr_event->station + 1].push(ter_event); //only for itself, so no need to send/recv
					}
					else
					{
						Init_paras ho;
						ho.id = HandOver; ho.time = curr_event->time + travelledDuration; ho.station = curr_event->station + 1;
						ho.duration = curr_event->duration - travelledDuration; ho.position = 0; ho.speed = curr_event->speed;
						MPI_Send(&ho,1,mpi_InitType,curr_event->station+2,0,MPI_COMM_WORLD);
					}
				}
				else
					droppedCalls[rank-1] += 1;
			}
			else
			{
				cout<<"default, no correct event id!"<<endl;
			}
			curr_time = curr_event->time;
			cout<<"process: "<<rank<<", curr time: "<<curr_time<<", curr id: "<<curr_event->id<<endl;
			localQue[rank].pop();

			//after execute the current event in the queue, check whether there is new event in buffer to be added into the queue
			MPI_Status status;
			MPI_Recv(in_buffer,1,mpi_InitType,rank-1,0,MPI_COMM_WORLD,&status);
			int count;
			MPI_Get_count(&status, mpi_InitType, &count);
			for(int d=0; d < count; d++)
			{
				if(in_buffer[d].id == Termination)
				{
					Event* new_event = new Event(in_buffer[d].time,in_buffer[d].station);
					localQue[rank].push(new_event);
				}
				else if(in_buffer[d].id == Initiation)
				{
					Event* new_event = new Event(in_buffer[d].time,in_buffer[d].speed,in_buffer[d].station,in_buffer[d].duration,in_buffer[d].position);
					localQue[rank].push(new_event);
				}
				else if(in_buffer[d].id == HandOver)
				{
					Event* new_event = new Event(in_buffer[d].time,in_buffer[d].speed,in_buffer[d].station,in_buffer[d].duration);
					localQue[rank].push(new_event);
				}
				//in_buffer[d] = NULL;
			}			
		}
	}		
	MPI_Finalize();
	return 0;
}

